package com.boot.consumer;

import com.boot.code.MqConstant;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

/**
 * 
 * @author threestrong
 * @since 2022-08-09 11:50:00
 * @desc 多队列-RoketMQ多消费者开启
 *
 */
@SuppressWarnings("deprecation")
@Component
public class RocketMessageMultiConsumer {

	private static Logger logger = LoggerFactory.getLogger(RocketMessageMultiConsumer.class);

	public RocketMessageMultiConsumer(){}

	/**
	 * 开启多消费者消费
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public void startConsumers() {
		try {
			String queue_name = MqConstant.TOPIC;
			// 1.创建⼀个消费者的客户端
			DefaultMQPushConsumer consumer = createConsumer(queue_name);
			// 2. 消费者订阅Topic和Tag
			consumer.subscribe(queue_name, "*");
			// 3.开启线程 push() API ⻓轮询消息队列
			new RocketConsumerTopicThread(consumer, queue_name).start();
		} catch (MQClientException mqClientException) {
			throw new RuntimeException(mqClientException);
		}
	}

	/**
	 * 
	 * @param group 消费者对应的group,这里默认为RocketMQ的topic
	 * @return 特定topic对应的消费者
	 */
	@SuppressWarnings({ "rawtypes"})
	private DefaultMQPushConsumer createConsumer(String group) {
		// 创建消费者实例
		DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group);
		// 设置NameServer地址
		consumer.setNamesrvAddr("192.168.229.130:9876");
		// 设置消息模型:集群模式
		consumer.setMessageModel(MessageModel.CLUSTERING);
		// 设置每次拉取最大消息数(默认为32)
		consumer.setPullBatchSize(500);
		// 设置每次消费最大消息数(默认为1)
		consumer.setConsumeMessageBatchMaxSize(500);
		// 设置最大重试次数
		consumer.setMaxReconsumeTimes(5);
		// 设置拉取消息的间隔时间(默认为0)
		consumer.setPullInterval(1000);
		// 设置消费超时时间(分钟)
		consumer.setConsumeTimeout(15);
		return consumer;
	}
}
